package sparkStreaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.junit.Test

class WindowDriver {

	@Test
	def window() {

		val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
		val ssc = new StreamingContext(conf, Seconds(3))
		ssc.checkpoint(".")

		// Create a DStream that will connect to hostname:port, like localhost:9999
		val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

		val windowLine: DStream[String] = lines.window(Seconds(9), Seconds(3))

		// Split each line into words
		val words = windowLine.flatMap(_.split(" "))

		//import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
		// Count each word in each batch
		val wordCounts = words.map(word => (word, 1)).reduceByKey(_+_)

		// Print the first ten elements of each RDD generated in this DStream to the console
		wordCounts.print()

		ssc.start()             // Start the computation
		ssc.awaitTermination()  // Wait for the computation to terminate
		//ssc.stop()
	}

	@Test
	def StateWindow() {

		// 定义更新状态方法，参数values为当前批次单词频度，state为以往批次单词频度
		val updateFunc = (values: Seq[Int], state: Option[Int]) => {
			val sum = values.sum
			val stateSum = state.getOrElse(0)
			Some(sum + stateSum)
		}

		val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
		val ssc = new StreamingContext(conf, Seconds(3))
		ssc.checkpoint(".")

		// Create a DStream that will connect to hostname:port, like localhost:9999
		val lines = ssc.socketTextStream("localhost", 9999)

		// Split each line into words
		val words = lines.flatMap(_.split(" "))

		//import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
		// Count each word in each batch
		val pairs = words.map(word => (word, 1))

		val wordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b),Seconds(12), Seconds(6))

		// Print the first ten elements of each RDD generated in this DStream to the console
		wordCounts.print()

		ssc.start()             // Start the computation
		ssc.awaitTermination()  // Wait for the computation to terminate
		//ssc.stop()
	}

}
